fix: dispatch map lookups with normalized keys and nondeterministic null-guarded children - #5867
dwsmith1983 wants to merge 40 commits into
Conversation
github-actions
Bot
added
bug
ff812cc to
689c563
Compare
…ull-guarded children Map lookups with float, collated or complex keys were declined by the serde without a dispatch fallback, so the whole projection fell back to Spark. Four array and map serdes reproduce NULL propagation with a guard that serializes the child twice, so a stateful child was evaluated twice natively and returned wrong answers. Mix the codegen dispatch fallback into the map lookups, and decline any nondeterministic child in size, array_append, arrays_zip and map_from_arrays through one shared gate so Spark's generated code evaluates it once. Closes apache#5580 Closes apache#5781
689c563 to
dfbf000
Compare
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
The comment above getSupportLevel says only the array operand sits under the NULL guard, but the item is inside the THEN branch too, and DataFusion's CaseExpr filters the batch before it evaluates THEN. So the item is evaluated only on the rows the guard selects.
That matters for array_append specifically, because it is the one expression in this group whose Spark codegen does not short-circuit. On 3.4 and 3.5 ArrayAppend.doGenCode emits leftGen.code + rightGen.code + ctx.nullSafeExec(left.nullable, leftGen.isNull) { ... }, so rightGen.code runs on every row even when the array is NULL. ElementAt and MapFromArrays go through nullSafeCodeGen and Size has one child, which is why the guard matches Spark for those three.
On your own 16 row table, SELECT _1, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), monotonically_increasing_id()) FROM test_array_append_nondet gives Spark [1,0], [1,2], [1,4] and so on for the even rows, because the counter advances on all 16. Comet evaluates the native monotonically_increasing_id only over the 8 filtered rows and gives [1,0], [1,1], [1,2] up to [1,7]. The same filtering also swallows an error the item would have raised. Under ANSI, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), 1 / (_1 - 1)) raises DIVIDE_BY_ZERO in Spark at _1 = 1 and returns cleanly in Comet.
The expect_native case below uses array(1), which is non-nullable, so the guard mask is all true and CaseExpr takes the branch that skips the filter entirely. That case passes whatever the serde does, which is why the fixture does not catch this. Would it make sense for NullGuardSupport to cover both children of ArrayAppend, and to change that fixture to a nullable operand with a stateful item so it fails without the change?
On sequencing, I have commented on #5875 pointing it here, since this PR carries the null-guard correctness work as well as the map lookup dispatch and that makes it the better base. Two things to settle before it lands though. #5875 measured this exact routing change and found lookup-only projections running 8.7 to 11.4 percent slower with the dispatcher on at 65536 rows, with no reproducible benefit at 1024 rows. Do you have numbers for the routing here? And #5854 teaches the native map builders LAST_WIN and null key rejection, which would remove the Incompatible branch this PR routes through the dispatcher for map_from_arrays, so that one needs an order agreed with @peterxcli.
Last thing is coverage for the routes this enables. map_from_arrays_dedup_policy.sql sets spark.comet.exec.scalaUDF.codegen.enabled=false and keeps asserting expect_fallback(mapKeyDedupPolicy), so the new LAST_WIN dispatch route is never exercised. Could those queries assert expect_dispatch(map_from_arrays) instead, with a separate dispatcher-off file if the fallback is still worth pinning? The four new nondeterministic fixtures also build every array inline over an int column, so no dispatched kernel in them reads an Arrow ListVector from a column. #5844 hit a real Spark 3.4.3 ColumnarArray.copy problem in exactly that shape, so an array<int> column in at least the size and map_from_arrays fixtures would be worth having. While you are in expressions.md, cardinality still says Native two rows above size, and Spark registers both names against Size.
…d-nondeterministic-children
…in the dispatched fixtures
…e guard filters its evaluation
Yes to both.
Done:
Added to both: the dispatched kernel now reads its arrays from columns as well as building them inline.
Not yet; I will run the same shape #5875 measured before this lands. On #5854, once native LAST_WIN lands the |
|
Triage note: #5875 also closes #5580, and mixes From reading both, this one looks like the superset: it also closes #5781 for the stateful child being serialized twice under the |
…aches the null guards
The nested null guards serialize each child a second time inside the
`map_from_arrays` call, so a stateful child advances independently in
each copy: the guard's copy sees every row while the constructor's copy
sees only the rows the guard selected. With
map_from_arrays(IF(monotonically_increasing_id() % 2 != 0, array(1), NULL), array(2))
over sixteen rows in one partition, Spark returns eight maps and Comet
returned four (apache#5781). Under LAST_WIN this case used to fall back for
the policy alone, so running the policy natively exposed it there.
Port `NullGuardSupport` from apache#5867 unchanged in name, reason and
position, so that PR rebases by dropping the hunk, and decline a
nondeterministic child in `CometMapFromArrays.getSupportLevel` as
`Unsupported`; the projection falls back to Spark, which evaluates the
child once. apache#5867 still routes the same decline through the JVM codegen
dispatcher and applies it to `size`, `array_append` and `arrays_zip`.
Cover it with the query above as a Scala test on a one-partition table
under LAST_WIN, the same query in `map_from_arrays_dedup_policy.sql`,
and `map_from_arrays_nondeterministic_child.sql` for the default policy,
which mirrors the fixture in apache#5867 with `expect_fallback` in place of
`expect_dispatch`.
Yes, happy for this PR to carry the map-lookup serde. The |
andygrove
added
the
run-all-spark-profiles
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for the update. I reran this locally and it is in good shape. 646 of 646 on Spark 4.1 and 637 of 637 on 3.4, the 9 canceled ones being the collation fixtures. I also neutralized NullGuardSupport.nondeterministicChild so it never declines, rebuilt, and reran the new fixtures. The three that are reachable on 4.1 fail with "Results do not match for query" rather than only a routing assertion, so they have real teeth. The array_append native case is no longer vacuous either.
I have applied run-all-spark-profiles. Two reasons. CI had never actually run here, every Comet CI run on the branch was sitting at action_required. And the pull request tier only runs Spark 4.1, so array_append_nondeterministic_child.sql could not run pre-merge at all with its MaxSparkVersion: 3.5 cap.
One more overlap to settle before this lands. #5844 also mixes CodegenDispatchFallback into CometMapFromArrays and rewrites map_from_arrays_dedup_policy.sql the same way, so it duplicates this pull request as well as #5875 and #5854. Are you and @LinSimon-901101 happy for this one to carry that too? Worth pulling one thing across either way. #5844 found that on Spark 3.4.3 the LAST_WIN fixture needs spark.sql.parquet.enableNestedColumnVectorizedReader=false because ColumnarArray.copy drops primitive-array NULLs (SPARK-48019), and this pull request now feeds array columns through the dispatcher in the same shape. My 3.4 run is green so I think your data avoids it, but it is worth knowing about.
On the benchmark, I do not think you need to re-measure. #5875 measured this exact routing and found DOUBLE-key lookup-only projections 8.7 to 11.4 percent slower at 65,536 rows, with mixed projections 10.5 to 12.7 percent faster. Could you cite those numbers here and say plainly that the lookup-only regression is accepted in exchange for keeping the projection in Comet? Right now that tradeoff is only recorded on the other pull request.
| // The item sits inside the guard's THEN branch, and DataFusion's CaseExpr evaluates that | ||
| // branch only on the rows the guard selects, while Spark's codegen evaluates the item on | ||
| // every row. A stateful item therefore drifts the same way a stateful array does. | ||
| override def getSupportLevel(expr: ArrayAppend): SupportLevel = |
There was a problem hiding this comment.
I filed #6086 for the ANSI gap in your "What this does not cover" section, and confirmed it on main at 5ca1499. With spark.sql.ansi.enabled=true, array_append(IF(_1 % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>)), 1 / (_1 - 1)) raises DIVIDE_BY_ZERO in Spark and returns cleanly in Comet. A control query in the same fixture confirms Comet's ANSI divide does raise on its own, so the test is not vacuous.
Could you link that issue from this comment? getSupportLevel reports Compatible() for the case, so without it nothing in the code or in the generated compatibility guide records the divergence.
While I was there I checked whether CometMapFromArrays has the mirror-image problem, since its guard is IsNotNull(left) AND IsNotNull(right) while Spark's nullSafeCodeGen short-circuits the values array. It does not reproduce. That one runs natively and matches Spark. The negative result is written up in the issue so nobody re-derives it.
There was a problem hiding this comment.
Could you link that issue from this comment?
Linked in 0d61cac, and the support level changed with it. With ANSI on and a nullable array, getSupportLevel now returns Incompatible with the #6086 note, so array_append runs through the codegen dispatcher by default and raises where Spark raises. The native guarded kernel only runs there under allowIncompatible, and the generated guide records the divergence as the opt-in consequence. With ANSI off, or a non-nullable array, nothing changes. A new array_append_ansi_null_array.sql fixture pins the dispatched raise and the native path for a non-nullable array. #6086 stays open for evaluating the item outside the guard, which would bring the native path back under ANSI.
| */ | ||
| private[serde] object NullGuardSupport { | ||
|
|
||
| val nondeterministicReason: String = |
There was a problem hiding this comment.
This renders in the generated compatibility guide right next to MapKeySupport's reasons, which are full sentences about what Spark does and what the native path does instead. This one is a lowercase fragment about how the serde is built, which reads oddly as a user-facing bullet. Would something like "Comet has no native path for a nondeterministic operand such as rand() or monotonically_increasing_id(), because the native NULL guard would evaluate it twice" sit better there?
There was a problem hiding this comment.
Would something like "Comet has no native path for a nondeterministic operand such as
rand()ormonotonically_increasing_id(), because the native NULL guard would evaluate it twice" sit better there?
Yes. That is the reason now, word for word apart from the code formatting, and since the string is shared it changes the bullet for all four serdes in 0d61cac.
|
|
||
| override def getUnsupportedReasons(): Seq[String] = Seq( | ||
| "Not all input data types are supported; falls back to Spark for unsupported types") | ||
| "Not all input data types are supported; unsupported types run through the JVM codegen " + |
There was a problem hiding this comment.
GenerateDocs already prints "The following cases have no native implementation and always run in the JVM using Spark's code-generated implementation (inside the Comet pipeline)" above these bullets whenever the serde mixes in CodegenDispatchFallback. So "unsupported types run through the JVM codegen dispatcher" now repeats its own header, and the bullet still never says which types. Could it name them instead, something like map, interval and variant element types?
There was a problem hiding this comment.
Could it name them instead, something like map, interval and variant element types?
Done in 0d61cac. The bullet now names what isTypeSupported declines: map, calendar, day-time and year-month interval, variant, TIME and user-defined element types, and a struct or inner array holding one of those. The dispatcher sentence is gone since the generated header already says it.
| SELECT _1, arrays_zip(array(2), IF(monotonically_increasing_id() % 2 = 0, array(1), CAST(NULL AS ARRAY<INT>))) AS z | ||
| FROM test_arrays_zip_nondet | ||
|
|
||
| -- A deterministic nullable child stays on the native guarded path. |
There was a problem hiding this comment.
"A deterministic nullable child stays on the native guarded path" is sitting above the expect_dispatch query for a non-nullable stateful child, and the expect_native query below it has no comment at all. Looks like the two got swapped.
Separately, size and map_from_arrays picked up column-fed dispatch cases in this round but arrays_zip did not, and it is the one whose dispatched kernel copies ListVectors into an array<struct<>>. Would you mind adding one query here over an array<int> column?
There was a problem hiding this comment.
Looks like the two got swapped.
They were. The nullable-child comment now sits above the expect_native query. The fixture also gained an arr array<int> column, NULL on every fourth row, and a dispatched arrays_zip query that reads both arrays from it, so the kernel copies list vectors from the input rather than building them inline. Flipping that query to expect_native fails the suite, so the sentinel is load-bearing. Both in 0d61cac.
…e arrays_zip types With ANSI on, an item that raises on a row whose array is NULL raises in Spark but not on the native path, because the NULL guard skips the item there. The serde now reports that case as incompatible when the array is nullable, so it runs through the JVM codegen dispatcher by default and the native guard stays behind allowIncompatible. The divergence is recorded in the compatibility guide with its tracking issue, and a fixture pins the raise, the dispatch and the native path for a non-nullable array. The shared nondeterministic reason is now a full sentence for the generated guide, the arrays_zip type bullet names the element types the native kernel declines, and the arrays_zip fixture gets its comments in the right order plus a dispatched query over an array column.
Yes. This pull request already routes
Added to the description under a Performance heading, with the 65,536-row figures from #5875 and the tradeoff stated as you put it. The four inline points are addressed in 0d61cac, with a reply on each thread. The SQL file suite passes on Spark 3.5 with the new column-fed |
…d-nondeterministic-children
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Thanks for working through the last round. Everything I raised is addressed, and reporting the ANSI nullable-array case as Incompatible is a better answer than only linking #6086, because it keeps Compatible() honest.
One thing in the map fixtures before this merges. CAST(-0.0 AS DOUBLE) casts a decimal literal, and a decimal has no signed zero, so it produces +0.0. The lookups at lines 64, 67 and 76 of element_at_map.sql and line 50 of get_map_value.sql therefore look up +0.0, and the comments above them describe a -0.0 lookup that never happens. Could those use -0.0D instead, and CAST(-0.0D AS FLOAT) for the float case, so the fixtures exercise the normalization they describe?
An unsuffixed -0.0 is a decimal literal with no signed zero, so CAST(-0.0 AS DOUBLE) looked up +0.0 and found the key on the native path too. The lookups now use -0.0D and CAST(-0.0D AS FLOAT), which keep the sign, so the fixtures fail on results when the float key decline is missing. The contributor guide names the literal form as well.
Done. The four lookups now use The old form was hiding the answer check as well as the comment. With the float decline in |
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
…d-nondeterministic-children
2 participants
Which issue does this PR close?
Closes #5580, closes #5781.
Rationale for this change
Two serde gaps in the array and map expressions, both resolved by the codegen dispatcher rather than by native changes.
map_col[key]andelement_at(map, key)decline float, collated and complex map keys because the native lookup compares raw Arrow values where Spark normalizes-0.0, treats NaN as equal to itself, compares strings by collation and compares complex keys with interpreted ordering. The declines are right, but neither serde mixed inCodegenDispatchFallback, so the whole projection fell back to Spark instead of running Spark's own generated code inside the Comet pipeline.size,array_append,arrays_zipandmap_from_arraysreproduce Spark's NULL propagation with aCASE WHEN child IS NOT NULLguard that serializes the child twice. A stateful child advances each copy independently, so the guard and the operation see different rows and the answer is silently wrong: on a 16-row table withIF(monotonically_increasing_id() % 2 = 0, array(1), NULL)as the operand,sizereturned -1 on five rows where Spark returns 1,arrays_zipreturned[null, 2]for[1, 2],array_appendreturned[2]for[1, 2]andmap_from_arraysreturned NULL for{1 -> 2}.element_athad the same shape and was fixed in #5766 for its ANSI arm; these four are not ANSI-gated, so the wrong answers were reachable in every configuration.What changes are included in this PR?
CometMapExtractandCometElementAtmix inCodegenDispatchFallback, so the declined key types run through the dispatcher.CometElementAt's ANSI arm for a nondeterministic operand dispatches the same way.NullGuardSupportgate declines any nondeterministic child inCometSize,CometArrayAppend(the array operand only; the item is not under the guard),CometArraysZipandCometMapFromArrays, and all four mix inCodegenDispatchFallback, so Spark's generated code evaluates the child once. Nullability is not consulted: a non-nullable stateful child only stays correct today because DataFusion skips the filter when the guard matches every row, which is not a contract to rely on.getUnsupportedReasonslists updated for the doc generator, and the six affected rows in the expressions guide move from Native to Hybrid.How are these changes tested?
SQL-file fixtures over parquet tables, all asserting Spark's answer and the execution path:
*_nondeterministic_child.sqlfixtures, one per serde, with the stateful operand (dispatched), a non-nullable stateful operand (dispatched), and a deterministic nullable operand (native).array_appendalso pins that a stateful item stays native, and its fixture is capped at Spark 3.5 because 4.x rewritesarray_appendtoarray_insert. Before the change, the stateful cases failed as result mismatches with the values above.element_at_map.sqlandget_map_value.sqlflip their fallback cases to dispatch and add NaN lookups and a struct-keyed map column with a per-row key, a NULL inside the key and a NULL key.element_at_map_collation.sqlflips the same way and passes on the Spark 4.0 profile.element_at_ansi.sqlpins the dispatched nondeterministic arm.map_from_arrays_dedup_policy.sqldisables the dispatcher so its LAST_WIN fallback assertion keeps meaning what it says, matching the existingmap_from_entriesfixture.CometMapExpressionSuiterenames five fallback tests to dispatch tests through the helper that checks the dispatch tag.array_append_ansi_null_array.sqlruns with ANSI on and pins that an item raising on a NULL-array row raises through the dispatcher as in Spark, that a nullable array is dispatched, and that a non-nullable array stays native. Thearrays_zipfixture adds a dispatched query over anarray<int>column so the kernel copies list vectors from real input.578 of 578 across
CometSqlFileTestSuite,CometArrayExpressionSuiteandCometMapExpressionSuiteon Spark 3.5, the map fixtures on the Spark 4.0 profile, andtest-compileon 4.0.Performance
#5875 measured this same routing with
CometCodegenDispatchBenchmark. At 65,536 rows, DOUBLE-key lookup-only projections ran 8.7 to 11.4 percent slower through the dispatcher than with it switched off, and mixed projections with a DOUBLE key and a small map ran 10.5 to 12.7 percent faster. The slower lookup-only case is accepted here in exchange for keeping the whole projection in Comet, where before it fell back to Spark.What this does not cover
array_appendkeeps its item inside the native NULL guard'sTHENbranch, which DataFusion evaluates only on the rows where the array is not NULL, while Spark's codegen evaluates the item on every row. A nondeterministic item is now declined to the dispatcher for that reason. A deterministic item that raises under ANSI on a row whose array is NULL, such as1 / (_1 - 1), would return cleanly on the native path where Spark raises. With ANSI on and a nullable array the serde now reportsIncompatible, soarray_appendruns through the codegen dispatcher by default and raises like Spark, and the native path is available underallowIncompatiblewith that divergence recorded in the compatibility guide. Evaluating the item outside the guard, so the native path can serve ANSI too, is tracked in #6086.